/**
 * 
 */
package com.rrd.dw.mr.text;

import java.io.IOException;
import java.io.StringReader;
import java.lang.reflect.Field;
import java.util.UUID;

import javax.xml.bind.JAXBContext;
import javax.xml.bind.JAXBException;
import javax.xml.bind.Unmarshaller;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import com.rrd.dw.mr.text.model.BeCalllistStatus;
import com.rrd.dw.mr.text.model.BeContactLendStatus;
import com.rrd.dw.mr.text.model.BePhonelistStatus;
import com.rrd.dw.mr.text.model.ContactLendRequest;
import com.rrd.dw.mr.text.model.ContactbeContactLendStatus;
import com.rrd.dw.mr.text.model.Entend_info;
import com.rrd.dw.mr.text.model.HitRule;
import com.rrd.dw.mr.text.model.PBCCreditReportQueryRecord;
import com.rrd.dw.mr.text.model.PBCReportItem;
import com.rrd.dw.mr.text.model.Plist;
import com.rrd.dw.mr.text.model.PreCArequest;
import com.rrd.dw.mr.text.model.QHRisk;
import com.rrd.dw.mr.text.model.Zhima_credit_watch_detail;
import com.rrd.dw.utils.HadoopUtils;
import com.rrd.dw.utils.OutPutUtil;

/**
 * @author CQ
 */
public class RRDRiskAnalysisReportXMLParser extends Configured implements Tool {
    private static final String PBCReport = "PBCReport";
    private static final String BigTable = "BigTable";
    private static final String PBCReportItem = "PBCReportItem";
    private static final String PBCCreditReportQueryRecord = "PBCCreditReportQueryRecord";
    private static final String hitRules = "hitRules";
    private static final String Plist = "Plist";
    private static final String ZMCreditWatchlistdetails = "ZMCreditWatchlistdetails";
    private static final String ZMCreditWatchlistdetailsExtend = "ZMCreditWatchlistdetailsExtend";
    private static final String ContactLendRequest = "ContactLendRequest";
    private static final String beContactLendStatus = "beContactLendStatus";
    private static final String bePhonelistStatus = "bePhonelistStatus";
    private static final String beCalllistStatus = "beCalllistStatus";
    private static final String ContactbeContactLendStatus = "ContactbeContactLendStatus";

    public static class RRDAnalysisReportMapper
            extends Mapper<LongWritable, Text, Text, Text> {

        protected void map(LongWritable key, Text value, Context context)
                throws IOException, InterruptedException {
            String[] columns = value.toString().split(OutPutUtil.FILED_SPLIT,
                -1);
            if (columns.length != 40) {
                return;
            }
            String userKey = columns[1];
            String applicationId = columns[2];
            String input = columns[34];
            String step = columns[39];
            if (input != null && !"".equals(input.trim())) {

            }
            context.write(new Text(userKey + OutPutUtil.FILED_SPLIT
                + applicationId + OutPutUtil.FILED_SPLIT + step),
                new Text(input));
        }

    }

    public static class RRDAnalysisReportReducer
            extends Reducer<Text, Text, NullWritable, Text> {
        private MultipleOutputs<NullWritable, Text> mos = null;
        private StringReader sr;
        private JAXBContext jaxbContext;
        private Unmarshaller unmarshaller;

        protected void setup(Context context)
                throws IOException, InterruptedException {
            mos = new MultipleOutputs<NullWritable, Text>(context);
            try {
                jaxbContext = JAXBContext.newInstance(PreCArequest.class);
                unmarshaller = jaxbContext.createUnmarshaller();
            } catch (JAXBException e) {
                e.printStackTrace();
            }
        }

        protected void cleanup(Context context)
                throws IOException, InterruptedException {
            mos.close();
            jaxbContext = null;
            unmarshaller = null;
        }

        public void reduce(Text key, Iterable<Text> values, Context context)
                throws IOException, InterruptedException {

            String keyStr = key.toString() + OutPutUtil.FILED_SPLIT;
            for (Text val : values) {
                sr = new StringReader(val.toString());
                PreCArequest preCArequest = null;
                try {
                    preCArequest = (PreCArequest) unmarshaller.unmarshal(sr);
                } catch (JAXBException e) {
                    e.printStackTrace();
                }
                if (preCArequest == null) {
                    continue;
                }
                mos.write(RRDRiskAnalysisReportXMLParser.PBCReport,
                    NullWritable.get(),
                    new Text(keyStr + preCArequest.getPbcReport().toString()));
                StringBuffer bigtableSB = new StringBuffer();
                bigtableSB.append(preCArequest.getLendRequest().toString()
                    + OutPutUtil.FILED_SPLIT);
                bigtableSB.append(preCArequest.getTongDun().toString()
                    + OutPutUtil.FILED_SPLIT);
                bigtableSB.append(preCArequest.getBlackList().toString()
                    + OutPutUtil.FILED_SPLIT);
                bigtableSB.append(preCArequest.getJuXinLiMiGuan().toString()
                    + OutPutUtil.FILED_SPLIT);
                bigtableSB
                    .append(preCArequest.getBlacklist_category().toString()
                        + OutPutUtil.FILED_SPLIT);
                bigtableSB.append(preCArequest.getIspInfo().toString()
                    + OutPutUtil.FILED_SPLIT);
                bigtableSB.append(preCArequest.getIspContactList().toString()
                    + OutPutUtil.FILED_SPLIT);
                if (preCArequest.getQhRisks() != null
                    && preCArequest.getQhRisks().size() > 0) {
                    bigtableSB
                        .append(preCArequest.getQhRisks().get(0).toString()
                            + OutPutUtil.FILED_SPLIT);
                } else {
                    bigtableSB.append(
                        new QHRisk().toString() + OutPutUtil.FILED_SPLIT);
                }
                bigtableSB.append(preCArequest.getInvolvedInfo().toString()
                    + OutPutUtil.FILED_SPLIT);
                bigtableSB.append(preCArequest.getCcxPhone().toString()
                    + OutPutUtil.FILED_SPLIT);
                bigtableSB.append(preCArequest.getCcxPhoneStatus().toString()
                    + OutPutUtil.FILED_SPLIT);
                bigtableSB.append(preCArequest.getPengYuanEdu().toString()
                    + OutPutUtil.FILED_SPLIT);
                bigtableSB.append(preCArequest.getFacePlusPlus().toString()
                    + OutPutUtil.FILED_SPLIT);
                bigtableSB.append(preCArequest.getLendRequestCount().toString()
                    + OutPutUtil.FILED_SPLIT);
                bigtableSB.append(preCArequest.getAntiFraud().toString()
                    + OutPutUtil.FILED_SPLIT);
                bigtableSB.append(
                    preCArequest.getUser_call_behavior_analysis().toString()
                        + OutPutUtil.FILED_SPLIT);
                bigtableSB.append(preCArequest.getReloan().toString()
                    + OutPutUtil.FILED_SPLIT);
                bigtableSB
                    .append(preCArequest.getZmCreditAntifraudScore().toString()
                        + OutPutUtil.FILED_SPLIT);
                bigtableSB.append(preCArequest.getZmCreditAntifraudVerifyCode()
                    + OutPutUtil.FILED_SPLIT);
                bigtableSB.append(
                    preCArequest.getRisk_codes() + OutPutUtil.FILED_SPLIT);
                bigtableSB.append(preCArequest.getZmCreditScore().toString()
                    + OutPutUtil.FILED_SPLIT);
                bigtableSB
                    .append(preCArequest.getZmCreditWatchlistii().toString()
                        + OutPutUtil.FILED_SPLIT);
                bigtableSB
                    .append(preCArequest.getExternal_blacklist().toString()
                        + OutPutUtil.FILED_SPLIT);
                bigtableSB.append(preCArequest.getChannel().toString()
                    + OutPutUtil.FILED_SPLIT);
                bigtableSB.append(preCArequest.getSuanHua().toString()
                    + OutPutUtil.FILED_SPLIT);
                bigtableSB.append(preCArequest.getPhoneList().toString()
                    + OutPutUtil.FILED_SPLIT);
                bigtableSB.append(preCArequest.getTianJixgscore().toString());
                mos.write(RRDRiskAnalysisReportXMLParser.BigTable,
                    NullWritable.get(),
                    new Text(keyStr + bigtableSB.toString()));

                for (PBCReportItem item : preCArequest.getPbcReportItems()) {
                    mos.write(RRDRiskAnalysisReportXMLParser.PBCReportItem,
                        NullWritable.get(), new Text(keyStr + item.toString()));
                }
                for (PBCCreditReportQueryRecord record : preCArequest
                    .getPbcCreditReportQueryRecords()) {
                    mos.write(
                        RRDRiskAnalysisReportXMLParser.PBCCreditReportQueryRecord,
                        NullWritable.get(),
                        new Text(keyStr + record.toString()));
                }
                for (HitRule rule : preCArequest.getTongDun().getHitRules()) {
                    mos.write(RRDRiskAnalysisReportXMLParser.hitRules,
                        NullWritable.get(), new Text(keyStr + rule.toString()));
                }
                for (Plist plist : preCArequest.getPlists()) {
                    mos.write(RRDRiskAnalysisReportXMLParser.Plist,
                        NullWritable.get(),
                        new Text(keyStr + plist.toString()));
                }

                if (preCArequest.getZmCreditWatchlistii()
                    .getZhima_credit_watch_detail_list() != null) {
                    for (Zhima_credit_watch_detail detail : preCArequest
                        .getZmCreditWatchlistii()
                        .getZhima_credit_watch_detail_list()) {
                        String uuid = UUID.randomUUID().toString();
                        mos.write(
                            RRDRiskAnalysisReportXMLParser.ZMCreditWatchlistdetails,
                            NullWritable.get(), new Text(keyStr + uuid
                                + OutPutUtil.FILED_SPLIT + detail.toString()));
                        for (Entend_info extend : detail
                            .getEntend_info_list()) {
                            mos.write(
                                RRDRiskAnalysisReportXMLParser.ZMCreditWatchlistdetailsExtend,
                                NullWritable.get(),
                                new Text(keyStr + uuid + OutPutUtil.FILED_SPLIT
                                    + extend.toString()));
                        }

                    }
                }
                if (preCArequest.getContactLendRequestList() != null) {
                    for (ContactLendRequest request : preCArequest
                        .getContactLendRequestList()) {
                        mos.write(
                            RRDRiskAnalysisReportXMLParser.ContactLendRequest,
                            NullWritable.get(),
                            new Text(keyStr + request.toString()));
                    }
                }

                if (preCArequest.getBeContactLendStatusList() != null) {
                    for (BeContactLendStatus status : preCArequest
                        .getBeContactLendStatusList()) {
                        mos.write(
                            RRDRiskAnalysisReportXMLParser.beContactLendStatus,
                            NullWritable.get(),
                            new Text(keyStr + status.toString()));
                    }
                }
                if (preCArequest.getBePhonelistStatusList() != null) {
                    for (BePhonelistStatus status : preCArequest
                        .getBePhonelistStatusList()) {
                        mos.write(
                            RRDRiskAnalysisReportXMLParser.bePhonelistStatus,
                            NullWritable.get(),
                            new Text(keyStr + status.toString()));
                    }
                }
                if (preCArequest.getBeCalllistStatusList() != null) {
                    for (BeCalllistStatus status : preCArequest
                        .getBeCalllistStatusList()) {
                        mos.write(
                            RRDRiskAnalysisReportXMLParser.beCalllistStatus,
                            NullWritable.get(),
                            new Text(keyStr + status.toString()));
                    }
                }
                if (preCArequest.getContactbeContactLendStatusList() != null) {
                    for (ContactbeContactLendStatus status : preCArequest
                        .getContactbeContactLendStatusList()) {
                        mos.write(
                            RRDRiskAnalysisReportXMLParser.ContactbeContactLendStatus,
                            NullWritable.get(),
                            new Text(keyStr + status.toString()));
                    }
                }
                preCArequest = null;

            }

        }
    }

    @Override
    public int run(String[] args) throws Exception {
        if (args.length != 2) {
            System.err.println(
                "Usage: RRDRiskAnalysisReportXMLParser <intputDir> <outputDir>");
            return -1;
        }
        Configuration conf = getConf();
        Job job = Job.getInstance(conf);

        System.setProperty("HADOOP_USER_NAME", "hdfs");
        String inputDir = args[0];
//        String outputDir = "/data/fdm/rrd02_risk_analysis_report_datas/";
        String outputDir = args[1];

        HadoopUtils.delete(conf, new Path(outputDir));

        job.setJarByClass(this.getClass());

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        job.setMapperClass(RRDAnalysisReportMapper.class);
        job.setReducerClass(RRDAnalysisReportReducer.class);

        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);

        FileInputFormat.setInputPaths(job, new Path(inputDir));
        FileOutputFormat.setOutputPath(job, new Path(outputDir));

        Field[] fields = this.getClass().getDeclaredFields();
        for (Field field : fields) {
            try {
                field.setAccessible(true);
                MultipleOutputs.addNamedOutput(job, field.get(this).toString(),
                    TextOutputFormat.class, NullWritable.class, Text.class);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        job.setNumReduceTasks(2);

        return job.waitForCompletion(true) ? 0 : 1;
    }

    /**
     * @param args
     * @throws Exception
     */
    public static void main(String[] args) throws Exception {
        System.exit(ToolRunner.run(new RRDRiskAnalysisReportXMLParser(), args));
    }
}
